<?php
/**
 * @Author: laoweizhen <1149243551@qq.com>,
 * @Date: 2021/10/27 18:52,
 * @LastEditTime: 2021/10/27 18:52
 */

namespace Lwz\LaravelExtend\MQ\Library\RocketMQ;

use App\Basic\Models\MqErrorLog;
use Illuminate\Support\Facades\Cache;
use Illuminate\Support\Facades\DB;
use Illuminate\Support\Facades\Log;
use Lwz\LaravelExtend\MQ\Constants\MQConst;
use Lwz\LaravelExtend\MQ\Enum\MQStatusLogEnum;
use Lwz\LaravelExtend\MQ\Events\MQExceptionEvent;
use Lwz\LaravelExtend\MQ\Exceptions\MQException;
use Lwz\LaravelExtend\MQ\Interfaces\MQConsumerInterface;
use Lwz\LaravelExtend\MQ\Interfaces\MQReliableConsumerInterface;
use Lwz\LaravelExtend\MQ\Interfaces\MQReliableProducerInterface;
use Lwz\LaravelExtend\MQ\Interfaces\MQStatusLogServiceInterface;
use Lwz\LaravelExtend\MQ\Interfaces\MQErrorLogServiceInterface;
use Lwz\LaravelExtend\MQ\Library\MQHelper;
use MQ\Exception\MessageNotExistException;
use MQ\Model\Message;
use MQ\MQConsumer;
use MQ\Exception\AckMessageException;

class RocketReliableConsumer implements MQReliableConsumerInterface
{
    use CommonTrait;

    /**
     * 分组id
     * @var string
     */
    protected string $groupId;

    /**
     * 消息标签
     * @var string|null
     */
    protected ?string $msgTag;

    /**
     * topic组
     * @var string|array
     */
    protected $topicGroup;

    /**
     * 消费组
     * @var string
     */
    protected string $consumeGroup;

    /**
     * 每次消费的消息数量(最多可设置为16条)
     * @var int
     */
    protected int $msgNum;

    /**
     * 长轮询时间（最多可设置为30秒）
     * @var int
     */
    protected int $waitSeconds;

    /**
     * 消息处理对象
     * @var MQConsumerInterface
     */
    protected MQConsumerInterface $mqHandleObj;

    /**
     * 消费者对象
     * @var MQConsumer
     */
    protected MQConsumer $consumer;

    /**
     * 是否保存消息失败日志
     * @var bool
     */
    protected bool $saveErrorLog;

    /**
     * RocketReliableConsumer constructor.
     * @param string|array $topicGroup topic组
     * @param int $msgNum 每次消费的消息数量(最多可设置为16条)
     * @param int $waitSeconds 长轮询时间（最多可设置为30秒）
     */
    public function __construct($topicGroup, string $consumeGroup, int $msgNum = 3, int $waitSeconds = 3)
    {
        // 更改日志驱动
//        Log::setDefaultDriver(config('mq.log_driver'));
        // 设置基本信息
        $this->_setMQInfo($topicGroup);

        // 设置消费组信息
        $this->setConsumeGroupInfo($consumeGroup);

        // 消费相关配置
        $this->topicGroup = $topicGroup;
        $this->consumeGroup = $consumeGroup;
        $this->msgNum = $msgNum;
        $this->waitSeconds = $waitSeconds;

        $this->saveErrorLog = config('mq.save_consumer_error_log');

        $this->setConsumer();
    }

    /**
     * 设置消费组信息
     * @param string $consumeGroup 消费组
     */
    protected function setConsumeGroupInfo(string $consumeGroup)
    {
        $cgInfo = config('mq.rocketmq.consume_group.' . $consumeGroup);
        // 验证消费组id（MQ的组件必填项）
        if (!isset($cgInfo['group_id']) || empty($cgInfo['group_id'])) {
            throw new MQException('请配置消费组的group_id');
        }

        // 验证消息处理类
        if (empty($cgInfo) || empty($handleClass = $cgInfo['handle_class'] ?? null)) {
            throw new MQException('请定义消息处理类');
        }
        $handleObj = new $handleClass;
        if (!$handleObj instanceof MQConsumerInterface) {
            throw new MQException('处理类必须实现 MQConsumerInterface 接口');
        }

        $msgTag = $cgInfo['msg_tag'] ?? null;
        $msgTagExt = $cgInfo['msg_tag_ext'] ?? null; // 如果配置了，优先使用这里的配置

        $this->mqHandleObj = $handleObj;
        $this->groupId = $this->setGroupIdExt($cgInfo['group_id']);
        $this->msgTag = MQHelper::setRocketMQMsgTagExt($msgTag, $msgTagExt);
    }

    /**
     * 设置消费者
     * @author lwz
     */
    private function setConsumer()
    {
        // 获取消费者（这里获取全部 msgTag 的消息）
        $this->consumer = RocketMQClient::getInstance()->getClient()->getConsumer($this->instanceId, $this->topic, $this->groupId, $this->msgTag);
    }

    /**
     * 消费
     * @return mixed
     * @throws MQException
     * @author lwz
     */
    public function consumer()
    {
        while (true) {
            try {
                // 长轮询消费消息
                // 如果topic没有消息则请求会在服务端挂住3s，3s内如果有消息可以消费则立即返回
                $messages = $this->consumer->consumeMessage($this->msgNum, $this->waitSeconds);
            } catch (\Exception $e) {
                if ($e instanceof MessageNotExistException) {
                    // Topic中没有消息可消费，继续轮询。
                    continue;
                }

                // 记录错误日志
                $this->saveErrorLog &&
                $this->getLogDriver()->info('获取消息 error:' . $e->getMessage());
                event(new MQExceptionEvent($e));
                sleep(3);
                continue;
            }

            /**
             * @var $message Message
             */
            // 处理业务逻辑
            $receiptHandles = array();
            foreach ($messages as $message) {
                // 消息句柄有时间戳，同一条消息每次消费拿到的都不一样
//                $receiptHandles[] = $message->getReceiptHandle();

                $msgTag = $message->getMessageTag(); // 消息标签
                $msgKey = $message->getMessageKey(); // 消息唯一标识
                $msgBody = $message->getMessageBody(); // 消息体
                $msgId = $message->getMessageId();

                // 格式化消息
                $msgBody = $this->decodeData($msgBody);
                try {

                    // 处理消息
                    // 参数一：获取不到 data 设置为 msgBody，主要为了兼容 之前版本
                    $this->mqHandleObj->handle($msgBody[MQConst::KEY_USER_DATA] ?? $msgBody, $msgKey, $msgTag);

                    // 消息句柄有时间戳，同一条消息每次消费拿到的都不一样
                    $receiptHandles[] = $message->getReceiptHandle();

                    // 如果处理消息没有抛出异常，则视为处理成功
                    // 如果配置了消费时删除日志，那么删除发送日志
                    $delSendLogState = $msgBody[MQConst::KEY_DELETE_SEND_LOG_STAGE] ?? null;
                    $delSendLogState == MQConst::DEL_SEND_LOG_CONSUMER && app(MQStatusLogServiceInterface::class)->deleteByMQUuid($msgKey);

                    // 消息确认
//                    $this->consumer->ackMessage([$message->getReceiptHandle()]);

                    // 消费重试消息，更新消息状态
                    $msgFrom = $msgBody[MQConst::KEY_MSG_FROM] ?? null;
                    $msgFrom == MQConst::MSG_FROM_CONSUME_ERROR_RETRY
                    && app(MQErrorLogServiceInterface::class)->updateByMQUuid($msgKey, ['retry_status' => MQStatusLogEnum::RES_CONSUME_SUCCESS]);

                    // 记录日志
                    config('mq.save_consumer_success_log')
                    && $this->getLogDriver()->info(
                        $this->_getLogMsg('[consumer success] 消费信息：', $msgId, $msgTag, $msgKey, $msgBody)
                    );

                } catch (\Throwable $throwable) {
                    // 消息确认
                    $receiptHandles[] = $message->getReceiptHandle();

                    $this->handleConsumeError($throwable, $msgId, $msgTag, $msgKey, $msgBody);
                }
            }

            try {
                // 消息确认
                $this->consumer->ackMessage($receiptHandles);
            } catch (\Exception $e) {
                if ($e instanceof AckMessageException) {
                    // 某些消息的句柄可能超时，会导致消息消费状态确认不成功。
                    $this->saveErrorLog
                    && $this->getLogDriver()->error("Ack Error, RequestId:%s\n", $e->getRequestId());
                    foreach ($e->getAckMessageErrorItems() as $errorItem) {
                        $this->saveErrorLog
                        && $this->getLogDriver()->error(sprintf("\tReceiptHandle:%s, ErrorCode:%s, ErrorMsg:%s\n", $errorItem->getReceiptHandle(), $errorItem->getErrorCode(), $errorItem->getErrorCode()));
                    }
                }
                event(new MQExceptionEvent($throwable));
            }
        }
    }

    /**
     * 处理消费异常
     * @param \Throwable $throwable
     * @param string $msgId
     * @param string $msgTag
     * @param string $msgKey
     * @param array $msgBody
     * @throws \Throwable
     * @author lwz
     */
    protected function handleConsumeError(\Throwable $throwable, string $msgId, string $msgTag, string $msgKey, array $msgBody)
    {
        // 记录日志
        $this->saveErrorLog
        && $this->getLogDriver()->error(
            $this->_getLogMsg('[consumer error] 消费信息：', $msgId, $msgTag, $msgKey, $msgBody)
        );

        $errorLogService = app(MQErrorLogServiceInterface::class);
        // 如果设置了重试次数，并且重试次数小于最大重试次数，就重试
        $maxRetryNum = config('mq.consume_error_max_retry', 3);

        $errLog = $errorLogService->getDetailByMquuid($msgKey);
        $retryNum = $errLog ? $errLog->retry_num : 0;
        $mqConfig = $this->_getMqBasicConfig();
        $errDesc = $this->getErrDesc($throwable);
        $retryStatus = $errLog ? $errLog->retry_status : MQStatusLogEnum::ERS_WAIT;
        $mqConfig[Constant::FIELD_TAG] = $msgTag;

        // 没有超过最大重试次数，重发数据
        if ($maxRetryNum && $retryNum < $maxRetryNum) {
            try {
                $this->mqReproduce($msgTag, $msgBody, $msgKey, $this->getRetryDelayTime($retryNum));
            } catch (\Throwable $t) {
                // 如果数据库没有错误日志，记录错误
                !$errLog && $errorLogService->addData($msgKey, $msgBody, $mqConfig, $errDesc, $retryNum, $retryStatus);
                throw $t;
            }
            $retryNum++;
            $retryStatus != MQStatusLogEnum::RES_CONSUME_SUCCESS && $retryStatus = MQStatusLogEnum::ERS_HAS_RETRY;
        }

        if ($errLog) {
            $errLog->retry_num = $retryNum;
            $errLog->payload = $msgBody;
            $errLog->retry_status = $retryStatus;
            $errLog->error_msg = $errDesc;
            $errLog->save();
        } else {
            $errorLogService->addData($msgKey, $msgBody, $mqConfig, $errDesc, $retryNum, $retryStatus);
        }

        // 错误处理
        event(new MQExceptionEvent($throwable));
    }

    /**
     * 获取延迟投递的时间
     * @param int $retryNum 重试次数
     * @return int
     * @author lwz
     */
    protected function getRetryDelayTime(int $retryNum): int
    {
        return time() + ($retryNum * 2 - 1) * 10;
    }

    /**
     * 获取日志消息
     * @param string $mainContent 主消息
     * @param string $msgId 消息id
     * @param string $msgTag 消息标签
     * @param string $msgKey 消息健
     * @param array $msgBody 消息体
     * @return string
     * @author lwz
     */
    private function _getLogMsg(string $mainContent, string $msgId, string $msgTag, string $msgKey, array $msgBody): string
    {
        return sprintf(
            $mainContent . '[msg_id] %s [msg_tag] %s; [msg_key] %s; [msg_body] %s',
            $msgId, $msgTag, $msgKey, MQHelper::encodeData($msgBody)
        );
    }
}
